from typing import Dict, Any, Optional, Union
import time
import re
from pydantic import BaseModel, Field, ValidationError
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from intelli_port.data_layer.clients import mysql_execute_read, mysql_execute_write, neo4j_exec, redis_client
from intelli_port.commons.audit import audit_event, write_ops_log
from intelli_port.langgraph_module.subgraphs_identity import build_identity_subgraph
from intelli_port.langgraph_module.subgraphs_payment import build_payment_subgraph
from intelli_port.langgraph_module.subgraphs_checkin import build_checkin_subgraph
from intelli_port.langgraph_module.subgraphs_rebook import build_rebook_subgraph
from intelli_port.langgraph_module.subgraphs_navigate import build_navigate_subgraph
from intelli_port.langgraph_module.subgraphs_baggage import build_baggage_subgraph
from intelli_port.langgraph_module.subgraphs_security import build_security_subgraph
from intelli_port.langgraph_module.subgraphs_assistance import build_assistance_subgraph
from intelli_port.commons.config import load_settings
import asyncio
from intelli_port.model.qwen.inferencer import create_qwen_inferencer
from intelli_port.rag.vector_rag import search_answer


def node_policy_guard(state: Dict[str, Any]) -> Dict[str, Any]:
    intent = state.get("intent")
    confidence = float(state.get("confidence") or 0)
    risk_flags = state.get("risk_flags") or {}
    try:
        s = load_settings()
        ov = ((state.get("session") or {}).get("ab") or {}).get("intent_low_conf_threshold")
        low_th = float(ov if ov is not None else getattr(s, "intent_low_conf_threshold", 0.5))
        if confidence < low_th:
            risk_flags["low_confidence"] = True
        slots = state.get("slots") or {}
        if intent == "pay":
            amt = float(slots.get("amount") or 0.0)
            if amt >= float(getattr(s, "pay_amount_confirm_threshold", 10000.0)):
                risk_flags["high_risk"] = True
        if intent == "refund":
            amt = float(slots.get("amount") or 0.0)
            if amt >= float(getattr(s, "refund_amount_confirm_threshold", 5000.0)):
                risk_flags["high_risk"] = True
    except Exception:
        pass
    return {**state, "risk_flags": risk_flags}


def _compiled():
    graph = StateGraph(dict)

    class IntentCatalogItem(BaseModel):
        name: str
        high_risk: bool = False
        required_slots: Optional[list[str]] = None

    INTENT_CATALOG: Dict[str, IntentCatalogItem] = {
        "gate_query": IntentCatalogItem(name="gate_query", high_risk=False, required_slots=["flight_no"]),
        "check_in": IntentCatalogItem(name="check_in", high_risk=True, required_slots=["ticket_no", "flight_date", "cabin_class"]),
        "rebook": IntentCatalogItem(name="rebook", high_risk=True, required_slots=["flight_no", "flight_date", "cabin_class"]),
        "refund": IntentCatalogItem(name="refund", high_risk=True, required_slots=["ticket_no"]),
        "pay": IntentCatalogItem(name="pay", high_risk=True, required_slots=["amount", "currency", "order_id"]),
        "identity": IntentCatalogItem(name="identity", high_risk=False, required_slots=[]),
        "faq": IntentCatalogItem(name="faq", high_risk=False, required_slots=[]),
        "navigate": IntentCatalogItem(name="navigate", high_risk=False, required_slots=["destination"]),
        "baggage": IntentCatalogItem(name="baggage", high_risk=False, required_slots=[]),
        "security_wait": IntentCatalogItem(name="security_wait", high_risk=False, required_slots=[]),
        "special_assistance": IntentCatalogItem(name="special_assistance", high_risk=False, required_slots=[]),
    }

    class GateQuerySlots(BaseModel):
        flight_no: Optional[str] = Field(default=None, description="航班号，如 CA1234")

    class CheckInSlots(BaseModel):
        ticket_no: Optional[str] = Field(default=None, description="票号，如 CA1234-001")
        flight_date: Optional[str] = Field(default=None, description="航班日期，YYYY-MM-DD 或中文日期")
        cabin_class: Optional[str] = Field(default=None, description="舱位枚举：F/C/Y")
        passengers: Optional[list] = Field(default=None, description="乘机人信息列表")

    class RebookSlots(BaseModel):
        flight_no: Optional[str] = Field(default=None, description="改签后航班号")
        flight_date: Optional[str] = Field(default=None, description="改签后航班日期")
        cabin_class: Optional[str] = Field(default=None, description="舱位枚举：F/C/Y")
        passengers: Optional[list] = Field(default=None, description="乘机人信息列表")
        order_id: Optional[str] = Field(default=None, description="订单号，可选")

    class NavigateSlots(BaseModel):
        origin: Optional[str] = Field(default=None, description="起点")
        destination: Optional[str] = Field(default=None, description="终点")
        gate: Optional[str] = Field(default=None, description="登机口")
        counter: Optional[str] = Field(default=None, description="值机柜台")

    class IdentityScanData(BaseModel):
        document_type: str = Field(description="证件类型：passport 或 id_card")
        mrz: Optional[str] = Field(default=None, description="护照MRZ字符串")
        id_no: Optional[str] = Field(default=None, description="身份证号码")

    class IntentOutput(BaseModel):
        intent: str = Field(description="意图标签")
        confidence: float = Field(ge=0, le=1, description="意图置信度")

    class SessionState(BaseModel):
        user_id: Optional[str] = None
        lang: Optional[str] = None
        last_intent: Optional[str] = None
        last_slots: Optional[Dict[str, Any]] = None
        risk_flags: Optional[Dict[str, Any]] = None
        device_id: Optional[str] = None
        location: Optional[str] = None

    class ServiceGateQuery(BaseModel):
        type: str = Field(default="gate_query")
        gate: str
        flight_no: Optional[str] = None

    class ServiceIdentity(BaseModel):
        type: str = Field(default="identity")
        verified: bool
        reason: Optional[str] = None

    class ServiceFAQ(BaseModel):
        type: str = Field(default="faq")
        answer: str

    class ServiceClarify(BaseModel):
        type: str = Field(default="clarify")
        prompt: str
        missing: Optional[list[str]] = None

    class ServiceCheckIn(BaseModel):
        type: str = Field(default="check_in")
        accepted: bool
        request_created: bool

    class ServicePay(BaseModel):
        type: str = Field(default="pay")
        payment_status: str
        txn_id: Optional[str] = None

    class ServiceRefund(BaseModel):
        type: str = Field(default="refund")
        refund_status: str
        refund_id: Optional[str] = None

    class ServiceRebook(BaseModel):
        type: str = Field(default="rebook")
        rebook_status: str
        order_id: Optional[str] = None

    ServiceUnion = Union[ServiceGateQuery, ServiceIdentity, ServiceFAQ, ServiceCheckIn, ServicePay, ServiceRefund, ServiceRebook]

    class ConfirmOutput(BaseModel):
        need_confirm: bool
        confirmed: Optional[bool] = None
        confirm_prompt: Optional[str] = None

    def node_preprocess(state: Dict[str, Any]) -> Dict[str, Any]:
        text = (state.get("text") or "").strip()
        scan_data = state.get("scan_data") or None
        modality = state.get("modality") or ("scan" if scan_data else "text")
        s = load_settings()
        sess_raw = state.get("session") or {}
        try:
            sess = SessionState.model_validate(sess_raw)
        except ValidationError:
            sess = SessionState()
        if not sess.lang:
            sess.lang = "zh"
        try:
            import hashlib, random
            uid = sess.user_id or ""
            dev = getattr(sess, "device_id", None) or ""
            base = (uid + "|" + dev + "|" + (text or ""))
            h = hashlib.sha1(base.encode("utf-8")).hexdigest()
            bucket_hash = int(h[:4], 16) % 100
            chosen_bucket = bucket_hash
            source = "hash"
            if getattr(s, "experiments_assign_enabled", False):
                # parse candidates from config or default to 0..99
                cands: list[int] = []
                try:
                    raw = (s.experiments_assign_candidates or "").strip()
                    if raw:
                        cands = [int(x) for x in re.split(r"[\s,]+", raw) if x.strip() != ""]
                    else:
                        cands = list(range(100))
                except Exception:
                    cands = list(range(100))
                try:
                    # score buckets using exposure/success counters (Beta sampling)
                    from intelli_port.commons.observability import ab_exposure_total
                    from intelli_port.api.routes_experiments import ab_success_total  # reuse counter instance
                    eid = "intent_weight_v1"
                    best_b = None
                    best_score = -1.0
                    for b in cands:
                        expo_b = 0.0
                        succ_b = 0.0
                        try:
                            from intelli_port.commons.observability import counter_sum_by_label
                            expo_b = counter_sum_by_label(ab_exposure_total, "bucket", str(b))
                            succ_b = counter_sum_by_label(ab_success_total, "bucket", str(b))
                        except Exception:
                            expo_b = 0.0
                            succ_b = 0.0
                        trials = max(1.0, float(expo_b))
                        alpha = 1.0 + float(succ_b)
                        beta = 1.0 + max(0.0, trials - float(succ_b))
                        score = random.betavariate(alpha, beta)
                        if score > best_score:
                            best_score = score
                            best_b = int(b)
                    if isinstance(best_b, int):
                        chosen_bucket = best_b
                        source = "online"
                except Exception:
                    chosen_bucket = bucket_hash
                    source = "hash"
            ab = {
                "experiment_id": "intent_weight_v1",
                "bucket": str(chosen_bucket),
            }
            if int(chosen_bucket) < 50:
                ab["intent_llm_weight"] = 0.7
                ab["intent_low_conf_threshold"] = 0.45
            else:
                ab["intent_llm_weight"] = 0.4
                ab["intent_low_conf_threshold"] = 0.55
            ab["source"] = source
            sd = sess.model_dump()
            sd["ab"] = ab
            try:
                from intelli_port.commons.observability import ab_exposure_total, ab_assignment_source_total
                ab_exposure_total.labels(experiment_id=ab["experiment_id"], bucket=ab["bucket"]).inc()
                ab_assignment_source_total.labels(experiment_id=ab["experiment_id"], bucket=ab["bucket"], source=source).inc()
            except Exception:
                pass
            sess = SessionState.model_validate(sd)
        except Exception:
            pass
        return {**state, "text": text, "scan_data": scan_data, "modality": modality, "session": sess.model_dump()}

    def _llm_intent(text: str, lang: str) -> Optional[Dict[str, Any]]:
        try:
            try:
                asyncio.get_running_loop()
                return None
            except RuntimeError:
                pass
            inf = create_qwen_inferencer()
            prompt = f"请仅输出JSON，字段 intent, confidence。文本：{text}"
            async def _run():
                return await inf.infer_short_text(prompt, lang)
            r = asyncio.run(_run())
            reply = str(r.get("reply", "")).strip()
            import json
            j = None
            try:
                j = json.loads(reply)
            except Exception:
                m = re.search(r"\{.*\}", reply, re.S)
                if m:
                    try:
                        j = json.loads(m.group(0))
                    except Exception:
                        j = None
            if not j:
                return None
            intent = str(j.get("intent") or "").strip()
            conf = float(j.get("confidence") or 0)
            if not intent:
                return None
            return {"intent": intent, "confidence": max(0.0, min(1.0, conf))}
        except Exception:
            return None

    def node_router(state: Dict[str, Any]) -> Dict[str, Any]:
        text = state.get("text", "")
        scan_data = state.get("scan_data")
        s = load_settings()
        sess = state.get("session") or {}
        lang = (sess or {}).get("lang") or "zh"
        if scan_data:
            intent = "identity"
            confidence = 1.0
        else:
            keywords_gate = ["登机口", "航班", "航班号", "gate", "flight"]
            keywords_checkin = ["值机", "check in", "check-in"]
            keywords_rebook = ["改签", "变更航班", "rebook"]
            keywords_refund = ["退票", "退款", "refund"]
            keywords_pay = ["支付", "付款", "pay"]
            keywords_nav = ["导航", "路线", "怎么去", "路径", "到", "route", "path", "navigate"]
            keywords_baggage = ["行李", "托运", "baggage", "luggage"]
            keywords_security = ["安检", "等待", "排队", "security"]
            keywords_assist = ["轮椅", "协助", "帮助", "assistance", "special assistance"]
            if any(k in text for k in keywords_nav):
                intent = "navigate"
                confidence = 0.8
            elif any(k in text for k in keywords_baggage):
                intent = "baggage"
                confidence = 0.75
            elif any(k in text for k in keywords_security):
                intent = "security_wait"
                confidence = 0.75
            elif any(k in text for k in keywords_assist):
                intent = "special_assistance"
                confidence = 0.8
            elif any(k in text.lower() for k in ["gate", "flight"]) or any(k in text for k in keywords_gate):
                intent = "gate_query"
                confidence = 0.9
            elif any(k in text.lower() for k in ["check in", "check-in"]) or any(k in text for k in keywords_checkin):
                intent = "check_in"
                confidence = 0.85
            elif any(k in text for k in keywords_rebook) or "rebook" in text.lower():
                intent = "rebook"
                confidence = 0.8
            elif any(k in text for k in keywords_refund) or "refund" in text.lower():
                intent = "refund"
                confidence = 0.8
            elif any(k in text for k in keywords_pay) or "pay" in text.lower():
                intent = "pay"
                confidence = 0.8
            else:
                intent = "faq"
                confidence = 0.6
        if s.enable_llm_intent:
            llm = _llm_intent(text, lang)
            if llm:
                sess = state.get("session") or {}
                iw_override = (sess or {}).get("ab", {}).get("intent_llm_weight") if isinstance((sess or {}).get("ab"), dict) else None
                iw_base = s.intent_llm_weight
                iw = max(0.0, min(1.0, float(iw_override if iw_override is not None else iw_base)))
                cr = confidence
                cl = float(llm.get("confidence") or 0.0)
                ir = intent
                il = str(llm.get("intent") or ir)
                sr = (1.0 - iw) * cr
                sl = iw * cl
                if il == ir:
                    intent = il
                    confidence = max(cr, cl)
                else:
                    if sl >= sr:
                        intent = il
                        confidence = max(cl, cr * 0.8)
        return {**state, "intent": intent, "confidence": confidence}

    def node_slot_filling(state: Dict[str, Any]) -> Dict[str, Any]:
        intent = state.get("intent")
        text = state.get("text", "")
        slots: Dict[str, Any] = state.get("slots") or {}
        def _normalize_flight_no(t: str) -> Optional[str]:
            x = re.sub(r"\s|[-_]", "", t)
            x = x.upper()
            x = x.replace("０", "0").replace("１", "1").replace("２", "2").replace("３", "3").replace("４", "4").replace("５", "5").replace("６", "6").replace("７", "7").replace("８", "8").replace("９", "9")
            m1 = re.search(r"([A-Z]{2,3}\d{2,4})", x)
            return m1.group(1) if m1 else None
        def _edit_distance(a: str, b: str, max_cut: int = 2) -> int:
            la, lb = len(a), len(b)
            if abs(la - lb) > max_cut:
                return max_cut + 1
            dp = list(range(lb + 1))
            for i in range(1, la + 1):
                prev = dp[0]
                dp[0] = i
                for j in range(1, lb + 1):
                    tmp = dp[j]
                    cost = 0 if a[i - 1] == b[j - 1] else 1
                    dp[j] = min(dp[j] + 1, dp[j - 1] + 1, prev + cost)
                    prev = tmp
            return dp[lb]
        def _suggest_flights(prefix: str, target: str, limit: int = 10) -> list[str]:
            if not prefix:
                return []
            try:
                like = prefix + "%"
                rows = mysql_execute_read("SELECT id, flight_no, origin, destination FROM flights WHERE flight_no LIKE %s LIMIT %s", (like, limit), fetch="all").get("rows", [])
                cands = [(r[0], r[1], r[2], r[3]) for r in rows if r]
                if not cands:
                    return []
                import time as _t
                now_ts = int(_t.time())
                seven_days_ago = _t.strftime("%Y-%m-%d %H:%M:%S", _t.localtime(now_ts - 7 * 86400))
                def _popularity(fid: int) -> float:
                    rr = mysql_execute_read("SELECT COUNT(1) FROM flight_events WHERE flight_id=%s AND created_at>=%s AND event IN ('query','click','confirm')", (fid, seven_days_ago), fetch="one").get("rows", [])
                    cnt = int(rr[0][0]) if rr and rr[0] and rr[0][0] is not None else 0
                    return float(cnt)
                def _ctx_match(orig: str | None, dest: str | None) -> float:
                    t = (state.get("text") or "").upper()
                    m = re.findall(r"\b[A-Z]{3}\b", t)
                    a = m[0] if len(m) > 0 else None
                    b = m[1] if len(m) > 1 else None
                    s = 0.0
                    if a and orig and a == str(orig).upper():
                        s += 0.5
                    if b and dest and b == str(dest).upper():
                        s += 0.5
                    return s
                kv_w_sim = mysql_execute_read("SELECT value FROM kv_config WHERE `key`=%s", ("ab.intent.w_sim",), fetch="one").get("rows", [])
                kv_w_pop = mysql_execute_read("SELECT value FROM kv_config WHERE `key`=%s", ("ab.intent.w_pop",), fetch="one").get("rows", [])
                kv_w_ctx = mysql_execute_read("SELECT value FROM kv_config WHERE `key`=%s", ("ab.intent.w_ctx",), fetch="one").get("rows", [])
                w_sim = float(kv_w_sim[0][0]) if kv_w_sim and kv_w_sim[0] and kv_w_sim[0][0] else 0.6
                w_pop = float(kv_w_pop[0][0]) if kv_w_pop and kv_w_pop[0] and kv_w_pop[0][0] else 0.25
                w_ctx = float(kv_w_ctx[0][0]) if kv_w_ctx and kv_w_ctx[0] and kv_w_ctx[0][0] else 0.15
                pops = []
                for fid, fno, orig, dest in cands:
                    pops.append(_popularity(fid))
                max_pop = max(pops) if pops else 1.0
                min_pop = min(pops) if pops else 0.0
                def _norm_pop(v: float) -> float:
                    if max_pop == min_pop:
                        return 0.0
                    return (v - min_pop) / (max_pop - min_pop)
                scored = []
                for idx, (fid, fno, orig, dest) in enumerate(cands):
                    d = _edit_distance(target, fno, 2)
                    if d <= 2:
                        sim = max(0.0, 1.0 - float(d) / 2.0)
                        popn = _norm_pop(pops[idx])
                        ctxs = _ctx_match(orig, dest)
                        score = w_sim * sim + w_pop * popn + w_ctx * ctxs
                        scored.append((score, d, fno))
                if not scored:
                    return []
                scored.sort(key=lambda x: (-x[0], x[1], x[2]))
                return [c for _, _, c in scored[:5]]
            except Exception:
                try:
                    like = prefix + "%"
                    rows = mysql_execute_read("SELECT flight_no FROM flights WHERE flight_no LIKE %s LIMIT %s", (like, limit), fetch="all").get("rows", [])
                    cands = [r[0] for r in rows if r and r[0]]
                    scored = []
                    for c in cands:
                        d = _edit_distance(target, c, 2)
                        if d <= 2:
                            scored.append((d, c))
                    scored.sort(key=lambda x: (x[0], x[1]))
                    return [c for _, c in scored[:5]]
                except Exception:
                    return []
        def _parse_date(t: str) -> Optional[str]:
            m = re.search(r"(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})", t)
            if m:
                y, mo, d = m.groups()
                return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}"
            m2 = re.search(r"(\d{1,2})[-/.](\d{1,2})", t)
            if m2:
                import time as _t
                y = _t.localtime().tm_year
                mo, d = m2.groups()
                return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}"
            m3 = re.search(r"(\d{1,2})月(\d{1,2})日", t)
            if m3:
                import time as _t
                y = _t.localtime().tm_year
                mo, d = m3.groups()
                return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}"
            t2 = t.strip()
            if any(k in t2 for k in ["今天", "今日", "today"]):
                import time as _t
                tm = _t.localtime()
                return f"{tm.tm_year:04d}-{tm.tm_mon:02d}-{tm.tm_mday:02d}"
            if any(k in t2 for k in ["明天", "tomorrow"]):
                import time as _t
                tm = _t.localtime(int(_t.time()) + 86400)
                return f"{tm.tm_year:04d}-{tm.tm_mon:02d}-{tm.tm_mday:02d}"
            if any(k in t2 for k in ["后天"]):
                import time as _t
                tm = _t.localtime(int(_t.time()) + 2 * 86400)
                return f"{tm.tm_year:04d}-{tm.tm_mon:02d}-{tm.tm_mday:02d}"
            return None
        if intent == "gate_query":
            m = _normalize_flight_no(text)
            if m:
                slots["flight_no"] = m
        elif intent == "check_in":
            m = re.search(r"([A-Z]{2}\d{2,4}-\d{3})", re.sub(r"\s", "", text.upper()))
            if m:
                slots["ticket_no"] = m.group(1)
        elif intent == "navigate":
            m_gate = re.search(r"登机口\s*([A-Z]\d{1,3})", text)
            if m_gate:
                slots["destination"] = m_gate.group(1)
            m_counter = re.search(r"值机柜台\s*([A-Za-z0-9-]+)", text)
            if m_counter:
                slots["destination"] = m_counter.group(1)
            if any(k in text for k in ["无障碍", "轮椅", "accessible"]):
                slots["prefer_accessible"] = True
        elif intent == "special_assistance":
            if any(k in text for k in ["轮椅", "wheelchair"]):
                slots["assistance_type"] = "wheelchair"
            elif any(k in text for k in ["引导", "guide"]):
                slots["assistance_type"] = "guide"
        d = _parse_date(text)
        if d:
            slots["flight_date"] = d
        # 舱位解析
        if any(k in text for k in ["头等舱", "F舱"]) or re.search(r"\bF\b", text):
            slots["cabin_class"] = "F"
        elif any(k in text for k in ["商务舱", "公务舱", "C舱"]) or re.search(r"\bC\b", text):
            slots["cabin_class"] = "C"
        elif any(k in text for k in ["经济舱", "Y舱"]) or re.search(r"\bY\b", text):
            slots["cabin_class"] = "Y"
        # 通用金额与订单提取
        amt = None
        m_amt = re.search(r"(?:￥|¥|RMB|CNY)\s?(\d+(?:\.\d+)?)", text)
        if m_amt:
            amt = float(m_amt.group(1))
            slots["amount"] = amt
            slots["currency"] = "CNY"
        m_amt2 = re.search(r"(\d+(?:\.\d+)?)(?:元|人民币)", text)
        if not amt and m_amt2:
            amt = float(m_amt2.group(1))
            slots["amount"] = amt
            slots["currency"] = "CNY"
        m_order = re.search(r"订单\s*([A-Za-z0-9-]+)", text)
        if m_order:
            slots["order_id"] = m_order.group(1)
        try:
            if intent == "gate_query":
                slots_model = GateQuerySlots.model_validate(slots)
                fn = slots_model.flight_no
                if fn:
                    r = mysql_execute_read("SELECT id FROM flights WHERE flight_no=%s", (fn,), fetch="one").get("rows", [])
                    if not r:
                        slots["flight_no"] = None
                        cat = INTENT_CATALOG.get(intent)
                        miss = []
                        if cat and cat.required_slots:
                            for k in cat.required_slots:
                                if slots.get(k) in (None, ""):
                                    miss.append(k)
                        prefix = re.match(r"([A-Z]{2,3}\d{0,4})", fn)
                        pfx = (prefix.group(1) if prefix else fn[:3])
                        suggs = _suggest_flights(pfx, fn)
                        try:
                            from intelli_port.commons.observability import clarify_suggestions_total
                            clarify_suggestions_total.labels(intent=intent).inc()
                        except Exception:
                            pass
                        return {**state, "slots": slots, "missing_slots": miss, "suggestion": f"未找到航班 {fn}", "suggested_flights": suggs}
                    else:
                        try:
                            fid = int(r[0][0])
                            mysql_execute_write(
                                "INSERT INTO flight_events(flight_id, event, detail, created_at) VALUES(%s,'query',%s,NOW())",
                                (fid, (state.get("text") or "")),
                            )
                        except Exception:
                            pass
            elif intent == "check_in":
                slots_model = CheckInSlots.model_validate(slots)
                # 强约束：舱位合法
                cc = slots_model.cabin_class
                if cc and cc not in {"F", "C", "Y"}:
                    slots = {}
                    return {**state, "slots": slots}
            elif intent == "navigate":
                slots_model = NavigateSlots.model_validate(slots)
            elif intent == "special_assistance":
                slots_model = NavigateSlots.model_validate(slots)
            else:
                if intent == "rebook":
                    slots_model = RebookSlots.model_validate(slots)
                    if not (slots_model.flight_no and slots_model.flight_date):
                        slots = {}
                        return {**state, "slots": slots}
                else:
                    slots_model = GateQuerySlots.model_validate({})
            slots = slots_model.model_dump()
        except ValidationError:
            slots = {}
        cat = INTENT_CATALOG.get(intent)
        missing = []
        if cat and cat.required_slots:
            for k in cat.required_slots:
                if slots.get(k) in (None, ""):
                    missing.append(k)
        return {**state, "slots": slots, "missing_slots": missing}

    def node_clarify(state: Dict[str, Any]) -> Dict[str, Any]:
        intent = state.get("intent")
        miss: list[str] = state.get("missing_slots") or []
        sess = state.get("session") or {}
        lang = (sess or {}).get("lang") or "zh"
        suggestion = state.get("suggestion") or ""
        templates = {
            "gate_query": {
                "flight_no": {
                    "zh": "请提供航班号，如 CA1234；支持大小写、去空格与短横线",
                    "en": "Please provide flight number, e.g., CA1234",
                },
                "flight_date": {
                    "zh": "请提供航班日期，格式 YYYY-MM-DD，或 3月5日/今天/明天/后天/周五",
                    "en": "Please provide flight date (YYYY-MM-DD) or today/tomorrow",
                },
            },
            "check_in": {
                "ticket_no": {
                    "zh": "请提供票号，如 CA1234-001",
                    "en": "Please provide ticket number, e.g., CA1234-001",
                },
                "flight_date": {
                    "zh": "请提供航班日期，格式 YYYY-MM-DD 或 3月5日",
                    "en": "Please provide flight date (YYYY-MM-DD)",
                },
                "cabin_class": {
                    "zh": "请提供舱位：F/C/Y",
                    "en": "Please provide cabin class: F/C/Y",
                },
            },
            "rebook": {
                "flight_no": {
                    "zh": "请提供改签后航班号，如 CA1234",
                    "en": "Please provide target flight number",
                },
                "flight_date": {
                    "zh": "请提供改签后日期，格式 YYYY-MM-DD",
                    "en": "Please provide target flight date",
                },
                "cabin_class": {
                    "zh": "请提供舱位：F/C/Y",
                    "en": "Please provide cabin class: F/C/Y",
                },
            },
            "refund": {
                "ticket_no": {
                    "zh": "请提供票号，如 CA1234-001",
                    "en": "Please provide ticket number",
                },
            },
            "pay": {
                "amount": {
                    "zh": "请提供金额，如 ￥123.45 或 123元",
                    "en": "Please provide amount",
                },
                "currency": {
                    "zh": "请提供币种，如 CNY",
                    "en": "Please provide currency",
                },
                "order_id": {
                    "zh": "请提供订单号",
                    "en": "Please provide order id",
                },
            },
        }
        def _label(k: str, l: str) -> str:
            m = {
                "zh": {
                    "flight_no": "航班号",
                    "flight_date": "航班日期",
                    "cabin_class": "舱位",
                    "ticket_no": "票号",
                    "amount": "金额",
                    "currency": "币种",
                    "order_id": "订单号",
                },
                "en": {
                    "flight_no": "flight number",
                    "flight_date": "flight date",
                    "cabin_class": "cabin",
                    "ticket_no": "ticket number",
                    "amount": "amount",
                    "currency": "currency",
                    "order_id": "order id",
                },
            }
            d = m.get(l) or m["zh"]
            return d.get(k, k)
        if lang == "en":
            if miss:
                base = "Please provide to continue " + intent + ": " + ",".join([_label(x, lang) for x in miss])
                adds = []
                tpls = templates.get(intent, {})
                for k in miss:
                    v = (tpls.get(k) or {}).get("en")
                    if v:
                        adds.append(v)
                suggs = state.get("suggested_flights") or []
                sug_text = ("; Candidates: " + ",".join(suggs)) if suggs else ""
                prompt = base + ("; " + "; ".join(adds) if adds else "") + (f"; Suggestion: {suggestion}" if suggestion else "") + sug_text
            else:
                prompt = "Please clarify your request or provide more information."
        else:
            if miss:
                base = "请补充以下信息以继续" + intent + "：" + ",".join([_label(x, lang) for x in miss])
                adds = []
                tpls = templates.get(intent, {})
                for k in miss:
                    v = (tpls.get(k) or {}).get("zh")
                    if v:
                        adds.append(v)
                suggs = state.get("suggested_flights") or []
                sug_text = ("；可能候选：" + ",".join(suggs)) if suggs else ""
                prompt = base + ("；" + "；".join(adds) if adds else "") + (f"；建议：{suggestion}" if suggestion else "") + sug_text
            else:
                prompt = "请澄清您的需求或提供更多信息。"
        try:
            from intelli_port.commons.observability import clarify_prompts_total
            clarify_prompts_total.labels(intent=intent).inc()
        except Exception:
            pass
        out = ServiceClarify(type="clarify", prompt=prompt, missing=miss).model_dump()
        return {**state, "service": out}

    def node_policy_guard2(state: Dict[str, Any]) -> Dict[str, Any]:
        intent = state.get("intent")
        confidence = float(state.get("confidence") or 0)
        risk_flags = state.get("risk_flags") or {}
        cat = INTENT_CATALOG.get(intent)
        if cat and cat.high_risk:
            risk_flags["high_risk"] = True
        s = load_settings()
        ov = ((state.get("session") or {}).get("ab") or {}).get("intent_low_conf_threshold")
        low_th = float(ov if ov is not None else getattr(s, "intent_low_conf_threshold", 0.5))
        if confidence < low_th:
            risk_flags["low_confidence"] = True
        slots = state.get("slots") or {}
        if intent == "pay":
            amt = float(slots.get("amount") or 0.0)
            if amt >= float(getattr(s, "pay_amount_confirm_threshold", 10000.0)):
                risk_flags["high_risk"] = True
        if intent == "refund":
            amt = float(slots.get("amount") or 0.0)
            if amt >= float(getattr(s, "refund_amount_confirm_threshold", 5000.0)):
                risk_flags["high_risk"] = True
        return {**state, "risk_flags": risk_flags}

    def node_policy_guard(state: Dict[str, Any]) -> Dict[str, Any]:
        intent = state.get("intent")
        confidence = float(state.get("confidence") or 0)
        risk_flags = state.get("risk_flags") or {}
        cat = INTENT_CATALOG.get(intent)
        if cat and cat.high_risk:
            risk_flags["high_risk"] = True
        if intent == "identity" and not state.get("scan_data"):
            risk_flags["identity_missing_scan"] = True
        # 设备标识缺失视为轻度风险
        try:
            sess0 = state.get("session") or {}
            if not (sess0.get("device_id") or "").strip():
                risk_flags["anonymous_device"] = True
        except Exception:
            pass
        s = load_settings()
        ov = ((state.get("session") or {}).get("ab") or {}).get("intent_low_conf_threshold")
        low_th = float(ov if ov is not None else getattr(s, "intent_low_conf_threshold", 0.5))
        if confidence < low_th:
            risk_flags["low_confidence"] = True
        slots = state.get("slots") or {}
        if intent == "pay":
            amt = float(slots.get("amount") or 0.0)
            if amt >= float(getattr(s, "pay_amount_confirm_threshold", 10000.0)):
                risk_flags["high_risk"] = True
        if intent == "refund":
            amt = float(slots.get("amount") or 0.0)
            if amt >= float(getattr(s, "refund_amount_confirm_threshold", 5000.0)):
                risk_flags["high_risk"] = True
        try:
            r = redis_client()
            sess = state.get("session") or {}
            uid = (sess or {}).get("user_id") or ""
            dev = (sess or {}).get("device_id") or ""
            now_s = int(time.time())
            scopes = [("per_min", 60, int(getattr(s, "rate_limits_per_min", 5))), ("per_10min", 600, int(getattr(s, "rate_limits_per_10min", 20))), ("per_hour", 3600, int(getattr(s, "rate_limits_per_hour", 60)))]
            for name, ttl, limit in scopes:
                if limit > 0:
                    k = f"rl:{name}:{uid}:{dev}:{intent}:{now_s // ttl}"
                    v = r.incr(k)
                    if v == 1:
                        r.expire(k, ttl)
                    if v > limit:
                        risk_flags["rate_limited"] = True
                        try:
                            from intelli_port.commons.observability import rate_limit_hits_total
                            rate_limit_hits_total.labels(scope=name).inc()
                        except Exception:
                            pass
                        break
        except Exception:
            pass
        try:
            bl = mysql_execute_read("SELECT account_id FROM risk_blacklist WHERE enabled=1 AND account_id=%s", ((state.get("session") or {}).get("user_id") or "",), fetch="one").get("rows", [])
            if bl:
                risk_flags["blacklisted"] = True
                try:
                    from intelli_port.commons.observability import blacklist_hits_total
                    blacklist_hits_total.labels(type="account").inc()
                except Exception:
                    pass
        except Exception:
            pass
        try:
            rs = 0.0
            if risk_flags.get("high_risk"):
                rs += 0.3
            if risk_flags.get("low_confidence"):
                rs += 0.3
            if risk_flags.get("rate_limited"):
                rs += 0.2
            if risk_flags.get("blacklisted"):
                rs += 0.5
            if risk_flags.get("identity_missing_scan"):
                rs += 0.3
            if risk_flags.get("anonymous_device"):
                rs += 0.1
            th = float(getattr(s, "risk_score_confirm_threshold", 0.7))
            risk_flags["risk_score"] = rs
            if rs >= th:
                risk_flags["high_risk"] = True
        except Exception:
            pass
        return {**state, "risk_flags": risk_flags}

    def node_confirm(state: Dict[str, Any]) -> Dict[str, Any]:
        intent = state.get("intent")
        confirm = bool(state.get("confirm") or False)
        rf = state.get("risk_flags", {})
        need = bool(rf.get("high_risk") or rf.get("rate_limited") or rf.get("blacklisted"))
        prompt = None
        sess = state.get("session") or {}
        lang = (sess or {}).get("lang") or "zh"
        ab = (sess or {}).get("ab") if isinstance((sess or {}).get("ab"), dict) else {}
        ab_eid = str((ab or {}).get("experiment_id") or "")
        ab_bkt = str((ab or {}).get("bucket") or "")
        if need and not confirm:
            s = state.get("slots") or {}
            if intent == "check_in":
                prompt = ("Confirm check-in, ticket: " + (s.get("ticket_no") or "-")) if lang == "en" else ("确认执行值机操作，票号：" + (s.get("ticket_no") or "-"))
            elif intent == "rebook":
                prompt = ("Confirm rebooking, flight: " + (s.get("flight_no") or "-")) if lang == "en" else ("确认执行改签操作，航班号：" + (s.get("flight_no") or "-"))
            elif intent == "refund":
                prompt = ("Confirm refund, ticket: " + (s.get("ticket_no") or "-")) if lang == "en" else ("确认执行退票/退款操作，票号：" + (s.get("ticket_no") or "-"))
            elif intent == "pay":
                amt = s.get("amount")
                cur = s.get("currency") or "CNY"
                oid = s.get("order_id")
                prompt = ("Confirm payment, amount: " + ((str(amt) + " " + cur) if amt else "-") + ", order: " + (oid or "-")) if lang == "en" else ("确认执行支付操作，金额：" + (((str(amt) + " " + cur)) if amt else "-") + "，订单：" + (oid or "-"))
            if rf.get("rate_limited"):
                prompt = ("Rate limited, please try later") if lang == "en" else ("操作过于频繁，请稍后再试")
            if rf.get("blacklisted"):
                prompt = ("Account restricted") if lang == "en" else ("账户受限，已拒绝操作")
        out = ConfirmOutput(need_confirm=need, confirmed=True if (need and confirm) else (False if need else None), confirm_prompt=prompt)
        if need and not confirm:
            try:
                from intelli_port.commons.observability import confirm_prompts_total
                confirm_prompts_total.labels(intent=intent).inc()
            except Exception:
                pass
            try:
                if ab_eid and ab_bkt:
                    from intelli_port.commons.observability import ab_clarify_prompts_total
                    ab_clarify_prompts_total.labels(experiment_id=ab_eid, bucket=ab_bkt, intent=intent).inc()
            except Exception:
                pass
        if need and confirm:
            try:
                from intelli_port.commons.observability import confirm_clicks_total
                confirm_clicks_total.labels(intent=intent).inc()
            except Exception:
                pass
            try:
                if ab_eid and ab_bkt:
                    from intelli_port.commons.observability import ab_clarify_success_total
                    ab_clarify_success_total.labels(experiment_id=ab_eid, bucket=ab_bkt, intent=intent).inc()
            except Exception:
                pass
            # 写入航班确认事件（用于热门度统计）
            try:
                sslots = state.get("slots") or {}
                fn = sslots.get("flight_no")
                if fn:
                    rr = mysql_execute_read("SELECT id FROM flights WHERE flight_no=%s", (fn,), fetch="one").get("rows", [])
                    if rr and rr[0] and rr[0][0] is not None:
                        fid = int(rr[0][0])
                        mysql_execute_write(
                            "INSERT INTO flight_events(flight_id, event, detail, created_at) VALUES(%s,'confirm',%s,NOW())",
                            (fid, intent),
                        )
            except Exception:
                pass
        if rf.get("blacklisted") or rf.get("rate_limited"):
            try:
                from intelli_port.commons.observability import risk_denials_total
                reason = "blacklisted" if rf.get("blacklisted") else "rate_limited"
                risk_denials_total.labels(intent=intent, reason=reason).inc()
            except Exception:
                pass
        return {**state, "confirm_output": out.model_dump()}

    def node_exec(state: Dict[str, Any]) -> Dict[str, Any]:
        intent = state.get("intent")
        slots = state.get("slots") or {}
        service: Dict[str, Any] = {}
        if intent == "gate_query":
            gate = None
            if slots.get("flight_no"):
                r = mysql_execute_read("SELECT g.code FROM flights f LEFT JOIN gates g ON f.gate_id=g.id WHERE f.flight_no=%s", (slots["flight_no"],), fetch="one").get("rows", [])
                if r:
                    gate = r[0][0]
            if not gate:
                try:
                    ng = neo4j_exec("MATCH (t:Terminal)-[:HAS_GATE]->(g:Gate {code:$c}) RETURN g", {"c": "A12"})
                    if ng and len(ng) > 0:
                        gate = ng[0].get("g", {}).get("code") or "A12"
                except Exception:
                    gate = None
            if not gate:
                try:
                    ra = search_answer(state.get("text", ""))
                    gate = ra.get("answer") or "A12"
                except Exception:
                    gate = "A12"
            service = {"type": "gate_query", "gate": gate}
            if slots.get("flight_no"):
                service["flight_no"] = slots["flight_no"]
        elif intent == "pay":
            pass
        elif intent == "refund":
            pass
        elif intent == "check_in":
            # 交由子图处理
            pass
        elif intent == "rebook":
            # 交由子图处理
            pass
        elif intent == "identity":
            raw_scan = state.get("scan_data") or {}
            try:
                scan = IdentityScanData.model_validate(raw_scan)
            except ValidationError:
                scan = IdentityScanData(document_type="unknown")
            doc_type = scan.document_type
            mrz = scan.mrz
            verified = False
            reason = None
            if doc_type == "passport":
                if mrz and isinstance(mrz, str) and len(mrz) >= 44:
                    verified = True
                else:
                    reason = "invalid_mrz"
            elif doc_type == "id_card":
                id_no = scan.id_no
                if id_no and len(str(id_no)) >= 10:
                    verified = True
                else:
                    reason = "invalid_id"
            else:
                reason = "unknown_document"
            service = {"type": "identity", "verified": verified}
            if reason:
                service["reason"] = reason
        else:
            try:
                ra = search_answer(state.get("text", ""))
                ans = ra.get("answer") or "为您提供机场常见问题解答。"
            except Exception:
                ans = "为您提供机场常见问题解答。"
            service = {"type": "faq", "answer": ans}
        try:
            if intent == "gate_query":
                validated: ServiceUnion = ServiceGateQuery.model_validate(service)
            elif intent == "check_in":
                validated = ServiceCheckIn.model_validate(service)
            elif intent == "pay":
                validated = ServicePay.model_validate(service)
            elif intent == "refund":
                validated = ServiceRefund.model_validate(service)
            elif intent == "rebook":
                validated = ServiceRebook.model_validate(service)
            elif intent == "identity":
                validated = ServiceIdentity.model_validate(service)
            else:
                validated = ServiceFAQ.model_validate(service)
            service = validated.model_dump()
            try:
                from intelli_port.commons.observability import exec_success_total
                exec_success_total.labels(intent=intent).inc()
            except Exception:
                pass
            try:
                sess2 = state.get("session") or {}
                ab2 = (sess2 or {}).get("ab") if isinstance((sess2 or {}).get("ab"), dict) else {}
                ab_eid2 = str((ab2 or {}).get("experiment_id") or "")
                ab_bkt2 = str((ab2 or {}).get("bucket") or "")
                if ab_eid2 and ab_bkt2:
                    from intelli_port.commons.observability import ab_exec_success_total, ab_path_success_total
                    ab_exec_success_total.labels(experiment_id=ab_eid2, bucket=ab_bkt2, intent=intent).inc()
                    path = str(service.get("type") or intent or "unknown")
                    ab_path_success_total.labels(experiment_id=ab_eid2, bucket=ab_bkt2, path=path).inc()
            except Exception:
                pass
        except ValidationError:
            try:
                from intelli_port.commons.observability import exec_fail_total
                exec_fail_total.labels(intent=intent).inc()
            except Exception:
                pass
            try:
                sess2 = state.get("session") or {}
                ab2 = (sess2 or {}).get("ab") if isinstance((sess2 or {}).get("ab"), dict) else {}
                ab_eid2 = str((ab2 or {}).get("experiment_id") or "")
                ab_bkt2 = str((ab2 or {}).get("bucket") or "")
                if ab_eid2 and ab_bkt2:
                    from intelli_port.commons.observability import ab_exec_fail_total
                    ab_exec_fail_total.labels(experiment_id=ab_eid2, bucket=ab_bkt2, intent=intent).inc()
            except Exception:
                pass
            service = {"type": "faq", "answer": "输出校验失败，已切换为通用回答。"}
        return {**state, "service": service, "schemas": {
            "intent": IntentOutput.model_json_schema(),
            "slots_gate_query": GateQuerySlots.model_json_schema(),
            "slots_check_in": CheckInSlots.model_json_schema(),
            "slots_rebook": RebookSlots.model_json_schema(),
            "service_gate_query": ServiceGateQuery.model_json_schema(),
            "service_check_in": ServiceCheckIn.model_json_schema(),
            "service_pay": ServicePay.model_json_schema(),
            "service_refund": ServiceRefund.model_json_schema(),
            "service_rebook": ServiceRebook.model_json_schema(),
            "service_identity": ServiceIdentity.model_json_schema(),
            "service_faq": ServiceFAQ.model_json_schema(),
        }}

    def node_service(state: Dict[str, Any]) -> Dict[str, Any]:
        return state

    identity_app = build_identity_subgraph()
    payment_app = build_payment_subgraph()
    navigate_app = build_navigate_subgraph()
    baggage_app = build_baggage_subgraph()
    security_app = build_security_subgraph()
    assistance_app = build_assistance_subgraph()
    graph.add_node("preprocess", node_preprocess)
    graph.add_node("router", node_router)
    graph.add_node("slot", node_slot_filling)
    graph.add_node("policy", node_policy_guard2)
    graph.add_node("confirm", node_confirm)
    graph.add_node("identity_subgraph", identity_app)
    graph.add_node("payment_subgraph", payment_app)
    checkin_app = build_checkin_subgraph()
    rebook_app = build_rebook_subgraph()
    graph.add_node("checkin_subgraph", checkin_app)
    graph.add_node("rebook_subgraph", rebook_app)
    graph.add_node("exec", node_exec)
    graph.add_node("service", node_service)
    graph.add_node("navigate_subgraph", navigate_app)
    graph.add_node("baggage_subgraph", baggage_app)
    graph.add_node("security_subgraph", security_app)
    graph.add_node("assistance_subgraph", assistance_app)
    graph.add_edge("preprocess", "router")
    try:
        def _route_from_router(state: Dict[str, Any]):
            if state.get("intent") == "identity":
                return "identity_subgraph"
            return "slot"
        graph.add_conditional_edges("router", _route_from_router)
    except Exception:
        graph.add_edge("router", "slot")
    graph.add_node("clarify", node_clarify)
    try:
        def _from_slot(state: Dict[str, Any]):
            miss = state.get("missing_slots") or []
            intent = state.get("intent")
            if miss:
                # gate_query 允许无航班号直接进入执行（返回默认/推荐登机口）
                if intent == "gate_query" and set(miss) == {"flight_no"}:
                    return "policy"
                return "clarify"
            return "policy"
        graph.add_conditional_edges("slot", _from_slot)
    except Exception:
        graph.add_edge("slot", "policy")
    try:
        def _from_policy(state: Dict[str, Any]):
            rf = state.get("risk_flags") or {}
            if rf.get("high_risk"):
                return "confirm"
            if rf.get("low_confidence"):
                return "clarify"
            if state.get("intent") == "navigate":
                return "navigate_subgraph"
            if state.get("intent") == "baggage":
                return "baggage_subgraph"
            if state.get("intent") == "security_wait":
                return "security_subgraph"
            if state.get("intent") == "special_assistance":
                return "assistance_subgraph"
            return "exec"
        graph.add_conditional_edges("policy", _from_policy)
    except Exception:
        graph.add_edge("policy", "confirm")
    try:
        def _from_confirm(state: Dict[str, Any]):
            out = state.get("confirm_output") or {}
            if out.get("need_confirm") and not out.get("confirmed"):
                return "service"
            if state.get("intent") in {"pay", "refund"}:
                return "payment_subgraph"
            if state.get("intent") == "check_in":
                return "checkin_subgraph"
            if state.get("intent") == "rebook":
                return "rebook_subgraph"
            if state.get("intent") == "special_assistance":
                return "assistance_subgraph"
            return "exec"
        graph.add_conditional_edges("confirm", _from_confirm)
    except Exception:
        graph.add_edge("confirm", "exec")
    graph.add_edge("identity_subgraph", "service")
    graph.add_edge("exec", "service")
    graph.add_edge("payment_subgraph", "service")
    graph.add_edge("checkin_subgraph", "service")
    graph.add_edge("rebook_subgraph", "service")
    graph.add_edge("navigate_subgraph", "service")
    graph.add_edge("baggage_subgraph", "service")
    graph.add_edge("security_subgraph", "service")
    graph.add_edge("assistance_subgraph", "service")
    graph.add_edge("clarify", "service")
    graph.set_entry_point("preprocess")
    memory = MemorySaver()
    return graph.compile(checkpointer=memory)


_app = _compiled()


def build_main_graph():
    return _app


def run_intent(text: str, modality: str | None = None, scan_data: Dict[str, Any] | None = None) -> Dict[str, Any]:
    payload: Dict[str, Any] = {"text": text}
    if modality:
        payload["modality"] = modality
    if scan_data:
        payload["scan_data"] = scan_data
    state = _app.invoke(payload, config={"configurable": {"thread_id": "default"}})
    return {"intent": state.get("intent", ""), "confidence": state.get("confidence", 0), "state": state}


def run_service(text: str, modality: str | None = None, scan_data: Dict[str, Any] | None = None, confirm: bool | None = None) -> Dict[str, Any]:
    payload: Dict[str, Any] = {"text": text}
    if modality:
        payload["modality"] = modality
    if scan_data:
        payload["scan_data"] = scan_data
    if confirm is not None:
        payload["confirm"] = confirm
    state = _app.invoke(payload, config={"configurable": {"thread_id": "default"}})
    return {"service": state.get("service", {}), "state": state}